package com.jd.platform.gobrs.worker.netty.pusher;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Queues;
import com.jd.platform.gobrs.common.model.GCacheKeyModel;
import com.jd.platform.gobrs.common.model.KeyTraffic;
import com.jd.platform.gobrs.common.model.typeenum.MessageType;
import com.jd.platform.gobrs.common.tool.FastJsonUtils;
import com.jd.platform.gobrs.worker.netty.dashboard.DashboardHolder;
import com.jd.platform.gobrs.worker.tool.AsyncPool;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 将热key推送到dashboard供入库
 * @author sizegang
 * @version 1.0
 * @date 2020-08-31
 */
@Component
public class DashboardPusher implements IPusher {
    /**
     * 热key集中营
     */
    private static LinkedBlockingQueue<GCacheKeyModel> manualKeyStoreQueue = new LinkedBlockingQueue<>();

    private static LinkedBlockingQueue<KeyTraffic> autoKeyStoreQueue = new LinkedBlockingQueue<>();

    @Override
    public void push(GCacheKeyModel model) {
        manualKeyStoreQueue.offer(model);
    }

    @Override
    public void pushAutoKey(KeyTraffic model) {
        autoKeyStoreQueue.offer(model);
    }

    @Override
    public void remove(GCacheKeyModel model) {

    }

    @PostConstruct
    public void uploadToDashboard() {
        executeManualKey();
        executeAutoKey();

    }


    private void executeManualKey() {
        AsyncPool.asyncDo(() -> {
            while (true) {
                try {
                    //要么key达到1千个，要么达到1秒，就汇总上报给etcd一次
                    List<GCacheKeyModel> tempModels = new ArrayList<>();
                    Queues.drain(manualKeyStoreQueue, tempModels, 1000, 1, TimeUnit.SECONDS);
                    if (CollectionUtil.isEmpty(tempModels)) {
                        continue;
                    }

                    //将热key推到dashboard
                    DashboardHolder.flushToDashboard(FastJsonUtils.convertObjectToJSON(tempModels), MessageType.REQUEST_HOT_KEY);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }


    private void executeAutoKey() {
        AsyncPool.asyncDo(() -> {
            while (true) {
                try {
                    //要么key达到1千个，要么达到1秒，就汇总上报给etcd一次
                    List<KeyTraffic> tempModels = new ArrayList<>();
                    Queues.drain(autoKeyStoreQueue, tempModels, 1000, 1, TimeUnit.SECONDS);
                    if (CollectionUtil.isEmpty(tempModels)) {
                        continue;
                    }
                    //将热key推到dashboard
                    DashboardHolder.flushToDashboard(FastJsonUtils.convertObjectToJSON(tempModels), MessageType.AUTO_REQUEST_HOT_KEY);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

}
